package join

import java.time.{Instant, LocalDateTime}

import ods.SourceKafka
import org.apache.flink.api.common.state.StateDescriptor.Type
import org.apache.flink.api.common.state.{ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.typeinfo.TypeHint
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

object BasedCoProcessDoubleStream {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

//    val stream1: KeyedStream[(String, Int), String] = env.fromElements(
//      ("o1",5000),
//      ("o4",20000),
//      ("o7",28000),
//      ("o2",150000)
//    ).assignAscendingTimestamps(x => x._2).keyBy(x => x._1)
//
//    val stream2: KeyedStream[(String, Int), String] = env.fromElements(
//      ("o2",14000),
//      ("o3",20000),
//      ("o6",25000),
//      ("o1",100000)
//    ).assignAscendingTimestamps(x => x._2).keyBy(x => x._1)

    val stream1 = env.addSource(SourceKafka.getSource("stream1")).map(x => (x.split(",")(0), x.split(",")(1).toInt)).assignAscendingTimestamps(x => x._2).keyBy(x => x._1)
    val stream2 = env.addSource(SourceKafka.getSource("stream2")).map(x => (x.split(",")(0), x.split(",")(1).toInt)).assignAscendingTimestamps(x => x._2).keyBy(x => x._1)

    val connected: DataStream[String] = stream1.connect(stream2).process(new KeyedCoProcessFunction[String, (String, Int), (String, Int), String] {
      lazy private val state1: ValueState[Int] = getRuntimeContext.getState(new ValueStateDescriptor[Int]("state1", Types.of[Int]))
      lazy private val state2: ValueState[Int] = getRuntimeContext.getState(new ValueStateDescriptor[Int]("state2", Types.of[Int]))

      override def processElement1(value: (String, Int), ctx: KeyedCoProcessFunction[String, (String, Int), (String, Int), String]#Context, out: Collector[String]): Unit = {
        if (state2.value() == 0) {
          state1.update(1)
          ctx.timerService().registerEventTimeTimer(value._2 + 5000)
        } else {
          state2.clear()
          out.collect(value._1 + "is ok.|" + System.currentTimeMillis())
        }
      }

      override def processElement2(value: (String, Int), ctx: KeyedCoProcessFunction[String, (String, Int), (String, Int), String]#Context, out: Collector[String]): Unit = {
        if (state1.value() == 0) {
          state2.update(1)
          ctx.timerService().registerEventTimeTimer(value._2 + 5000)
        } else {
          state1.clear()
          out.collect(value._1 + " is ok.|" + System.currentTimeMillis())
        }
      }

      override def onTimer(timestamp: Long, ctx: KeyedCoProcessFunction[String, (String, Int), (String, Int), String]#OnTimerContext, out: Collector[String]): Unit = {
        if (state1.value() != 0) {
          out.collect(ctx.getCurrentKey + " is not ok.1")
          state1.clear()
        }

        if (state2.value() != 0) {
          out.collect(ctx.getCurrentKey + " is not ok.2")
          state2.clear()
        }
      }
    })

    connected.print()

    env.execute()
  }
}
